#include "../include/my_mqtt.h"
#include <sys/ipc.h>
#include <sys/msg.h>
#include "../include/my_msq.h"
#include <string.h>

static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
static MQTTClient_message pubmsg = MQTTClient_message_initializer;
static MQTTClient client;
static MQTTClient_deliveryToken deliveredtoken, temptoken;
int count;

//互斥锁(publish用)
pthread_mutex_t mutex;

int my_mqtt_init(){
    //初始化互斥锁资源
    pthread_mutex_init(&mutex, NULL);

    //1.1 创建MQTT客户端
    deliveredtoken = 0;
    int retCode;
    if ((retCode = MQTTClient_create(&client, ADDRESS, CLIENTID,
        MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS)
    {
         printf("Failed to create client, return code [%d]\n", retCode);
         return -1;
    }
    //1.2 连接MQTT服务器
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    conn_opts.username = USERNAME;   //用户名
    conn_opts.password = PASSWORD;   //密码
    
    if ((retCode = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", retCode);
        return -1;
        
    }
    //1.3 订阅主题1
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n", 
        TOPIC_CMD_DEVICEID, CLIENTID, QOS);
        
    if ((retCode = MQTTClient_subscribe(client, TOPIC_CMD_DEVICEID, QOS)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to subscribe, return code %d\n", retCode);
        return -1;
    }
    //1.4 订阅主题2
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n", 
        TOPIC_CMD_WAVE, CLIENTID, QOS);
        
    if ((retCode = MQTTClient_subscribe(client, TOPIC_CMD_WAVE, QOS)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to subscribe, return code %d\n", retCode);
        return -1;
    }
    //
    return 0;
}

//订阅消息处理线程入口
void* my_mqtt_sub_thread(void* args){
    while(1){
        //printf("my_mqtt_sub_thread[running....]\n");
        int topicLen;
        int rc;
        char* topic = TOPIC_CMD_DEVICEID;
        MQTTClient_message *message = NULL;
        //装置id
        rc = MQTTClient_receive(client, &topic, &topicLen, &message, TIMEOUT);
        if(rc != MQTTCLIENT_SUCCESS || topic == NULL){
            continue;
        }
        printf("111: rc=[%d], topic=[%s]\n", rc, topic);
        if(strcmp(TOPIC_CMD_WAVE, topic) == 0){
            //处理 主题 cmde_wave 指令
            //TODO 写入消息队列 to 数据接收程序
            printf("收到触发波形的命令 \n");
            //
            my_msq_send_cmd();
            printf("已经往消息队列中写入了指令\n");
            continue;
        }
        if(strcmp(TOPIC_CMD_DEVICEID, topic) == 0){
            printf("收到 cmd_deviceid\n");
            //处理 主题 cmd_deviceid 指令
            pthread_mutex_lock(&mutex);
            //发布 data_deviceid主题
            publish_data_deviceid();
            pthread_mutex_unlock(&mutex);
            continue;
        }
        printf("不支持的主题信息 topic=[%s]\n", topic);
        sleep(1);
    }
}

//消息发布线程
void* my_mqtt_pub_thread(void* args){
    while(1){
        //阻塞式的从消息队列中收取数据
        unsigned int wave_data[1000];
        my_msq_rcv_wave(wave_data);
        printf("从消息队列中收到了波形数据？\n");
        //组包，通过MQTT发布主题
        cJSON *root = cJSON_CreateObject();
        if (root == NULL) {
            printf("创建JSON Object失败 [%s][%d]\n", __func__, __LINE__);
            continue;
        }

        unsigned int totalVal=0;
        unsigned int maxVal=wave_data[0];
        unsigned int minVal=wave_data[0];
        for(int i=0;i<1000;i++){
            unsigned int temp = wave_data[i];
            totalVal += temp;
            maxVal = maxVal<temp?temp:maxVal;
            minVal = minVal>temp?temp:minVal;
        }
        unsigned int avgVal= totalVal/1000;


        cJSON_AddStringToObject(root, "deviceid", CLIENTID);
        cJSON_AddStringToObject(root, "takeTime", "xxxx-xx-xx xx:xx:xx");
        cJSON_AddNumberToObject(root, "takeMs", 111);
        cJSON_AddNumberToObject(root, "maxValue", maxVal);
        cJSON_AddNumberToObject(root, "minValue", minVal);
        cJSON_AddNumberToObject(root, "avgValue", avgVal);
        cJSON_AddNumberToObject(root, "dataLen", 1000);
        //
        cJSON *arr = cJSON_CreateIntArray(wave_data, 1000);
        cJSON_AddItemToObject(root, "datas", arr);
        char *json_string = cJSON_Print(root);
        //发送数据
        //处理 主题 cmd_deviceid 指令
        pthread_mutex_lock(&mutex);
        publish_data_wave(json_string);
        pthread_mutex_unlock(&mutex);


        //回收资源
        cJSON_Delete(root);
        free(json_string);
    }
}


//发布主题
void publish_data_deviceid(){
    MQTTClient_deliveryToken token;
    int rc;
    cJSON *root = cJSON_CreateObject();
    if (root == NULL) {
        return;
    }

    cJSON_AddStringToObject(root, "deviceid", CLIENTID);
    char *json_string = cJSON_Print(root);
    //publish
    pubmsg.payload = json_string;
    pubmsg.payloadlen = (int)strlen(json_string);
    pubmsg.qos = QOS;
    pubmsg.retained = 1;

    if ((rc = MQTTClient_publishMessage(client, TOPIC_DATA_DEVICEID, &pubmsg, &token)) != MQTTCLIENT_SUCCESS)
    {
         printf("Failed to publish message, return code [%d], topic=[%s]\n", rc, TOPIC_DATA_DEVICEID);
    }

    cJSON_Delete(root);
    free(json_string);
}

//发布主题
void publish_data_wave(char* jsonStr){
    MQTTClient_deliveryToken token;
    int rc;

    //publish
    pubmsg.payload = jsonStr;
    pubmsg.payloadlen = (int)strlen(jsonStr);
    pubmsg.qos = QOS;
    pubmsg.retained = 1;

    if ((rc = MQTTClient_publishMessage(client, TOPIC_DATA_WAVE, &pubmsg, &token)) != MQTTCLIENT_SUCCESS)
    {
         printf("Failed to publish message, return code [%d], topic=[%s]\n", rc, TOPIC_DATA_DEVICEID);
    }
}